package finalProject;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URL;
import java.util.HashSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;


public class Step2 {

	public static class MapClass extends
	Mapper<NounPair, Text, NounPair, Text> {

		
		@Override
		protected void map(NounPair key, Text value, Context context)
				throws IOException, InterruptedException {

				context.write(key, value);
		}
	}

	public static class PartitionClass extends Partitioner<NounPair, Text> {

		@Override
		public int getPartition(NounPair key, Text value, int partitionNum) {

			int ans = (key.getNPx().charAt(0) * 31 ^ 7) % partitionNum;
			while (ans < 0) {
				ans += partitionNum;
			}
			return ans;
		}
	}

	public static class ReduceClass extends Reducer<NounPair, Text, NullWritable, Text>{
		private static Text eol = new Text("\n"); 
		
		static HashSet<NounPair> annotatedTrue ;
		static HashSet<NounPair> annotatedFalse ;		

		private MultipleOutputs<NullWritable, Text> mos;
		
		/* 
		 * read the annotated set to a static 
		 */
		@Override
		protected void setup(Context context) throws IOException,
		InterruptedException {
			readAnnotated();
			mos = new MultipleOutputs<NullWritable, Text>(context);
		}

		@Override
		protected void reduce(NounPair key, Iterable<Text> values,
				Context context)
						throws IOException, InterruptedException {
		
			String outputName;
			String hypernym = "";
			if(annotatedTrue.contains(key) == true || annotatedFalse.contains(key) == true)
			{
				outputName = "known";
				if(annotatedTrue.contains(key) == true)
				{
					hypernym = "True";
				}
				else
				{
					hypernym = "False";
				}
			}
			else
			{
				outputName = "unknown";
			}
			
			String line = key.toString();
			mos.write(outputName, NullWritable.get(), new Text(line));
			
			for (Text pattern : values) {
				mos.write(outputName, NullWritable.get(), pattern);
			}
			
			if (hypernym.equals(""))
			{
				line = hypernym;
			}
			else
			{
				line = hypernym + eol;
			}
			mos.write(outputName, NullWritable.get(), new Text(line));
		}
		
		
		
		@Override
		protected void cleanup(Context context) throws IOException,
				InterruptedException {
			
			mos.close();
		}

		private static void readAnnotated() throws IOException {
			annotatedTrue = new HashSet<NounPair>();
			annotatedFalse = new HashSet<NounPair>();
			URL filePate = new URL("http://www.cs.bgu.ac.il/~dsp142/wiki.files/hypernym.txt");
			BufferedReader in = new BufferedReader(new InputStreamReader(filePate.openStream()));

			Stemmer stemmer = new Stemmer();
			String inputLine;
			while ((inputLine = in.readLine()) != null)
			{
				String[] input = inputLine.split("\t");
				if(input.length>=3){
					String first, second;
					stemmer.add(input[0].toCharArray(), input[0].length());
					stemmer.stem();
					first = stemmer.toString();
					stemmer.add(input[1].toCharArray(), input[1].length());
					stemmer.stem();
					second = stemmer.toString();
					if(input[2].toLowerCase().equals("true") == true)
					{
						annotatedTrue.add(new NounPair(input[0], input[1]));
						annotatedTrue.add(new NounPair(first, second));
					}
					else
					{
						annotatedFalse.add(new NounPair(input[0], input[1]));
						annotatedFalse.add(new NounPair(first, second));
					}
				}
			}
			in.close();
		}

	}

	public static void main(String[] args) throws IOException,
	ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		Job job = new Job(conf, "Final");
		job.setJarByClass(Step2.class);
		job.setMapperClass(MapClass.class);
		job.setPartitionerClass(PartitionClass.class);
		job.setReducerClass(ReduceClass.class);
		job.setMapOutputKeyClass(NounPair.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(NullWritable.class);
		job.setOutputValueClass(Text.class);
		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		FileInputFormat.addInputPath(job, new Path("s3n://ahgfinal/output/Step1/"));
		FileOutputFormat.setOutputPath(job, new Path("s3n://ahgfinal/output/Step2/"));
		MultipleOutputs.addNamedOutput(job, "known", TextOutputFormat.class, NullWritable.class, Text.class);
        MultipleOutputs.addNamedOutput(job, "unknown", TextOutputFormat.class, NullWritable.class, Text.class);
		int completion = job.waitForCompletion(true) ? 0 : 1;
		System.exit(completion);
	}
}
